package drds.global_transaction.api.rpc.rpc_client.transaction_manager;

import drds.common.Constants;
import drds.global_transaction.api.global_transaction_id_context.GlobalTransactionIdContexts;
import drds.global_transaction.api.message_codec.HeartbeatMessage;
import drds.global_transaction.api.message_codec.Message;
import drds.global_transaction.api.message_codec.ResponseCode;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_begin.GlobalTransactionBeginRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_begin.GlobalTransactionBeginResponseMessage;
import drds.global_transaction.api.rpc.config.ClientConfig;
import drds.global_transaction.api.rpc.rpc_client.Key;
import drds.global_transaction.api.rpc.rpc_client.Role;
import drds.global_transaction.api.rpc.rpc_client.RpcClient;
import drds.global_transaction.api.rpc.rpc_client.SendRequestMessagesRunnable;
import drds.global_transaction.common.GlobalTransactionIds;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.EventExecutorGroup;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool.impl.GenericKeyedObjectPool.Config;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

import static drds.global_transaction.common.exception.FrameworkErrorCode.NoAvailableService;

@Slf4j
@Sharable
public final class TransactionManagerRpcClient extends RpcClient
{
	private static final int max_merge_send_thread = 1;
	private static final long keep_alive_time = Integer.MAX_VALUE;
	private static final int max_queue_size = 2000;
	private static final String merge_thread_prefix = "rpcMergeMessageSend";
	private static final int schedule_interval_mills = 5;
	private static volatile TransactionManagerRpcClient instance;

	public static TransactionManagerRpcClient getInstance()
	{
		if (null == instance)
		{
			synchronized (TransactionManagerRpcClient.class)
			{
				if (null == instance)
				{
					ClientConfig clientConfig = new ClientConfig();
					final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(clientConfig.getClientWorkerThreads(), clientConfig.getClientWorkerThreads(), keep_alive_time, TimeUnit.SECONDS, new LinkedBlockingQueue(max_queue_size), new NamedThreadFactory(clientConfig.getTmDispatchThreadPrefix(), clientConfig.getClientWorkerThreads()), new ThreadPoolExecutor.CallerRunsPolicy());
					instance = new TransactionManagerRpcClient(clientConfig, null, threadPoolExecutor);
				}
			}
		}
		return instance;
	}

	public static TransactionManagerRpcClient getInstance(String applicationId, String transactionServiceGroup)
	{
		TransactionManagerRpcClient transactionManagerRpcClient = getInstance();
		transactionManagerRpcClient.setApplicationId(applicationId);
		transactionManagerRpcClient.setTransactionServiceGroup(transactionServiceGroup);
		return transactionManagerRpcClient;
	}

	@Setter
	@Getter
	private volatile ConcurrentMap<String, Object> serverAddressToLockObjectMap = new ConcurrentHashMap<String, Object>();
	// serverAddressToChannelMap and serverAddressToKeyMap
	// 通过serverAddress保证Channel和Key一一对应
	@Setter
	@Getter
	private volatile ConcurrentMap<String, Channel> serverAddressToChannelMap = new ConcurrentHashMap<String, Channel>();

	@Setter
	@Getter
	private volatile ConcurrentMap<String, Key> serverAddressToKeyMap = new ConcurrentHashMap<String, Key>();
	//
	@Setter
	@Getter
	private volatile AtomicBoolean initialized = new AtomicBoolean(false);
	@Setter
	@Getter
	private volatile ClientConfig clientConfig;

	@Setter
	@Getter
	private String applicationId;

	@Setter
	@Getter
	private String transactionServiceGroup;

	private TransactionManagerRpcClient(ClientConfig clientConfig, EventExecutorGroup eventExecutorGroup, ThreadPoolExecutor messageExecutor)
	{
		super(clientConfig, eventExecutorGroup, messageExecutor);
		this.clientConfig = clientConfig;
	}

//
	@Override
	protected Role getRole()
	{
		return Role.transaction_manager_client;
	}

	@Override
	protected Config getConfig()
	{
		Config config = new Config();
		config.maxActive = clientConfig.getMaxPoolActive();
		config.minIdle = clientConfig.getMinPoolIdle();
		config.maxWait = clientConfig.getMaxAcquireConnMills();
		config.testOnBorrow = clientConfig.isPoolTestBorrow();
		config.testOnReturn = clientConfig.isPoolTestReturn();
		config.lifo = clientConfig.isPoolFifo();
		return config;
	}

	//

	@Override
	public void init()
	{
		if (initialized.compareAndSet(false, true))
		{
			init(schedule_interval_mills, schedule_interval_mills);
		}
	}

	//

	public void init(long healthCheckDelay, long healthCheckPeriod)
	{
		super.init();
		ExecutorService executorService = new ThreadPoolExecutor(max_merge_send_thread, max_merge_send_thread, keep_alive_time, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory(getThreadPrefix(merge_thread_prefix), max_merge_send_thread));
		executorService.submit(new SendRequestMessagesRunnable(this));
		scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					reConnect();
				} catch (Exception ignore)
				{
					log.error(ignore.getMessage());
				}
			}
		}, healthCheckDelay, healthCheckPeriod, TimeUnit.SECONDS);
	}

	private void reConnect()
	{
		for (String serverAddress : globalTransactionServiceGroupAddressManager.lookup(transactionServiceGroup))
		{
			try
			{
				// connect(serverAddress);
				TimeUnit.SECONDS.sleep(1);
			} catch (Exception e)
			{
				log.error(FrameworkErrorCode.NetConnect.errCode, "can not connect to " + serverAddress + " cause:" + e.getMessage());
			}
		}

	}

	//

	@Override
	protected Channel connect(String serverAddress)
	{
		Channel channel = serverAddressToChannelMap.get(serverAddress);
		if (null != channel)
		{
			channel = getExistAliveChannel(channel, serverAddress);
			if (null != channel)
			{
				return channel;
			}
		}
		serverAddressToLockObjectMap.putIfAbsent(serverAddress, new Object());
		Object lockObject = serverAddressToLockObjectMap.get(serverAddress);
		synchronized (lockObject)
		{
			channel = connectWithLock(serverAddress);
			serverAddressToChannelMap.put(serverAddress, channel);
			return channel;
		}
	}

	private Channel getExistAliveChannel(Channel channel, String serverAddress)
	{
		if (channel.isActive())
		{
			return channel;
		} else
		{
			int i = 0;
			for (; i < ClientConfig.getMaxCheckAliveRetry(); i++)
			{
				try
				{
					Thread.sleep(ClientConfig.getCheckAliveInternal());
				} catch (InterruptedException exx)
				{
					log.error(exx.getMessage());
				}
				channel = serverAddressToChannelMap.get(serverAddress);
				if (null == channel || channel.isActive())
				{
					return channel;
				}
			}
			if (i == ClientConfig.getMaxCheckAliveRetry())
			{
				log.warn("channel " + channel + " is not active after long wait, close it.");
				releaseChannel(serverAddress, channel);
				return null;
			}
		}
		return null;
	}

	private Channel connectWithLock(String serverAddress)
	{
		Channel channel = serverAddressToChannelMap.get(serverAddress);
		if (null != channel && channel.isActive())
		{
			return channel;
		}
		//
		try
		{
			RegisterTransactionManagerRequestMessage registerTransactionManagerRequestMessage = new RegisterTransactionManagerRequestMessage(applicationId, transactionServiceGroup);
			serverAddressToKeyMap.putIfAbsent(serverAddress, new Key(serverAddress, getRole(), registerTransactionManagerRequestMessage));
			channel = keyToChannelMap.borrowObject(serverAddressToKeyMap.get(serverAddress));
		} catch (Exception exx)
		{
			log.error("get channel from pool error.", exx);
			throw new FrameworkException("can not register transaction_manager_client,err:" + exx.getMessage());
		}
		return channel;
	}

	@Override
	public void returnChannel(String serverAddress, Channel channel)
	{
		if (null == channel)
		{
			return;
		}
		try
		{
			if (channel.equals(serverAddressToChannelMap.get(serverAddress)))
			{
				serverAddressToChannelMap.remove(serverAddress);
			}
			keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
		} catch (Exception exx)
		{
			log.error("return channel to rpcPool error:" + exx.getMessage());
		}
	}

	public void releaseChannel(String serverAddress, Channel channel)
	{
		if (null == channel || null == serverAddress)
		{
			return;
		}
		try
		{
			Object object = serverAddressToLockObjectMap.get(serverAddress);
			synchronized (object)
			{
				Channel $channel = serverAddressToChannelMap.get(serverAddress);
				if (null == $channel)
				{
					keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
					return;
				}
				if ($channel.compareTo(channel) == 0)
				{
					returnChannel(serverAddress, channel);
				} else
				{
					keyToChannelMap.returnObject(serverAddressToKeyMap.get(serverAddress), channel);
				}
			}
		} catch (Exception exx)
		{
			log.error(exx.getMessage());
		}
	}

	//
	//
	@Override
	public void userEventTriggered(ChannelHandlerContext channelHandlerContext, Object event)
	{
		if (event instanceof IdleStateEvent)
		{
			IdleStateEvent idleStateEvent = (IdleStateEvent) event;
			//
			if (idleStateEvent.state() == IdleState.READER_IDLE)
			{
				if (log.isInfoEnabled())
				{
					log.info("channel" + channelHandlerContext.channel() + " read idle.");
				}
				try
				{
					keyToChannelMap.invalidateObject(serverAddressToKeyMap.get(channelHandlerContext.channel().remoteAddress()), channelHandlerContext.channel());
				} catch (Exception exx)
				{
					log.error(exx.getMessage());
				} finally
				{
					releaseChannel(getRemoteAddress(channelHandlerContext), channelHandlerContext.channel());
				}
			}
			//
			if (idleStateEvent == IdleStateEvent.WRITER_IDLE_STATE_EVENT)
			{
				try
				{
					if (log.isInfoEnabled())
					{
						log.info("will send ping msg,channel" + channelHandlerContext.channel());
					}
					synchronousSendRequest(channelHandlerContext.channel(), HeartbeatMessage.ping);
				} catch (Throwable throwable)
				{
					log.error("", "send request error", throwable);
				}
			}
		}

	}

	@Override
	public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception
	{
		if (threadPoolExecutor.isShutdown())
		{
			return;
		}
		releaseChannel(Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress()), channelHandlerContext.channel());
		super.channelInactive(channelHandlerContext);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) throws Exception
	{
		log.error(FrameworkErrorCode.ExceptionCaught.errCode, Nets.getHostAddressAndPort(channelHandlerContext.channel().remoteAddress()) + "connect exception. " + cause.getMessage(), cause);
		//
		releaseChannel(getRemoteAddress(channelHandlerContext.channel()), channelHandlerContext.channel());
		log.error("remove exception resource_manager_client channel:" + channelHandlerContext.channel());
		//
		super.exceptionCaught(channelHandlerContext, cause);
	}

	//

	//
	@Override
	public void onRegisterMessageSuccess(String serverAddress, Channel channel, Message requestMessage, Object response)
	{
		if (null != serverAddressToChannelMap.get(serverAddress) && serverAddressToChannelMap.get(serverAddress).isActive())
		{
			return;
		}
		serverAddressToChannelMap.put(serverAddress, channel);
	}

	@Override
	public void onRegisterMessageFail(String serverAddress, Channel channel, Message requestMessage, Object response)
	{
		if (response instanceof RegisterTransactionManagerResponseMessage && log.isInfoEnabled())
		{
			log.info("register client failed, transaction_coordinator version:" + ((RegisterTransactionManagerResponseMessage) response).getVersion());
		}
		throw new FrameworkException("register client app failed.");
	}

	@Override
	public Object asynchronousSendRequestWithTimeoutGetResponse(String serverAddress, Object message, long timeout) throws TimeoutException
	{
		return asynchronousSendRequestWithTimeoutGetResponse(serverAddress, connect(serverAddress), message, timeout);
	}

	@Override
	public void synchronousSendResponse(String serverAddress, long messageId, Object message)
	{
		super.synchronousSendResponse(connect(serverAddress), messageId, message);
	}

	//
	@Override
	public Object asynchronousSendRequestWithTimeoutGetResponse(Object message) throws TimeoutException
	{
		return asynchronousSendRequestWithTimeoutGetResponse(message, ClientConfig.getRpcRequestTimeout());
	}

	@Override
	public Object asynchronousSendRequestWithTimeoutGetResponse(Object message, long timeout) throws TimeoutException
	{
		//
		String addressAndPort = GlobalTransactionIds.getAddressAndPort(GlobalTransactionIdContexts.getGlobalTransactionIdWithAddressAndPort());
		String serverAddress = addressAndPort != null ? addressAndPort : loadBalance();
		Channel channel = connect(serverAddress);
		if (Constants.developMode)
		{
		    String logString="";
            for (Channel $channel: serverAddressToChannelMap.values()) {
                logString="channel:" + $channel.remoteAddress().toString()+",id:"+$channel.id().asLongText() + ",isActive:" + $channel.isActive()+",localAddress:"+$channel.localAddress().toString();
                log.debug(logString);
            }
            logString="channel:" + channel.remoteAddress().toString()+",id:"+channel.id().asLongText() + ",isActive:" + channel.isActive()+",localAddress:"+channel.localAddress().toString();
            log.debug(logString);
		}
		Object result = super.asynchronousSendRequestWithTimeoutGetResponse(serverAddress, channel, message, timeout);
		if (result == null)
		{
			log.error("result is null");
			releaseChannel(serverAddress, channel);// 不再进行交互了
			throw new IllegalStateException();
		} else
		{
			log.info("response is " + result);
			log.info("result class type is :" + result.getClass().getName());
			/**
			 * 首次连接
			 */
			if (result instanceof GlobalTransactionBeginResponseMessage && ((GlobalTransactionBeginResponseMessage) result).getResponseCode() == ResponseCode.failed)
			{
				releaseChannel(serverAddress, channel);// 不再进行交互了
				log.error("globalTransactionBegin response error,release channel:" + channel);
			}
		}
		return result;
	}

	private String loadBalance()
	{
		String[] addresses = globalTransactionServiceGroupAddressManager.lookup(transactionServiceGroup);
		if (addresses == null || addresses.length == 0)
		{
			throw new FrameworkException(NoAvailableService);
		}
		// Just single transaction_coordinator node
		return addresses[0];
	}

//

}
